package com.xxx.springboot.service.impl;

import com.xxx.springboot.service.RedisDelayQueueService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

/**
 * 2. 【redis延时队列】添加/获取任务service接口实现类
 * https://www.toutiao.com/i6954587200172392971
 */
@Slf4j
@Service
public class RedisDelayQueueServiceImpl implements RedisDelayQueueService {

    @Autowired
    private RedissonClient redissonClient;

    @Override
    public <T> void addTaskIntoDelayQueue(T value, long delayTime, TimeUnit timeUnit, String queueName) {
        try {
            // 1.获取阻塞队列（实际使用了list的rpush操作）
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueName);
            // 2.获取延时队列
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            // 3.往队列里添加任务
            delayedQueue.offer(value, delayTime, timeUnit);
            log.info("添加任务到延时队列成功。 queueName:{}, value:{}, 延时时间:{}秒", queueName, value, timeUnit.toSeconds(delayTime));
        } catch (Exception e) {
            log.error("添加任务到延时队列失败", e);
            throw new RuntimeException("添加任务到延时队列失败");
        }
    }

    @Override
    public <T> T getTaskFromDelayQueue(String queueName) throws InterruptedException {
        // 获取阻塞队列
        RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueName);
        // 从队列里拿取任务
        // take()方法从阻塞队列中拿取数据，当队列为空时，线程阻塞，直到队列中有数据
//        T value = (T) blockingDeque.take(); // 线程会一直阻塞在这里
        // poll()方法从阻塞队列中拿取数据，当队列不为空时，等待指定时间后返回null
        T value = (T) blockingDeque.poll(10, TimeUnit.MILLISECONDS);
        if(value!=null){
            log.info("获取延时队列中的任务成功 value:{}", value);
        }
        return value;
    }
}
